package com.gitee.xhs.savePointReader.keyedState.listStateReader;

import com.gitee.xhs.keyedState.listState.ListStateConstant;
import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.state.api.functions.KeyedStateReaderFunction;
import org.apache.flink.util.Collector;

/**
 * @author BigHead
 * @date 2022/6/30 10:29
 * @description ListStateReaderFunction
 */
public class ListStateReaderFunction extends KeyedStateReaderFunction<Long, ListStateExhibitor> {

    private final String jobId;
    private final String operatorUid;
    private transient ListState<Transaction> listState;

    public ListStateReaderFunction(String jobId, String operatorUid) {
        this.jobId = jobId;
        this.operatorUid = operatorUid;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        listState = getRuntimeContext().getListState(ListStateConstant.LIST_STATE_DESCRIPTOR);
    }

    @Override
    public void readKey(Long key, Context ctx, Collector<ListStateExhibitor> out) throws Exception {
        ListStateExhibitor listStateExhibitor = new ListStateExhibitor();
        listStateExhibitor.setJobId(jobId);
        listStateExhibitor.setOperatorUid(operatorUid);
        listStateExhibitor.setKey(key);
        for (Transaction transaction : listState.get()) {
            listStateExhibitor.addStateValueList(transaction.getAmount());
        }
        out.collect(listStateExhibitor);
    }
}
